package cn.com.sparkle.paxos.addprocess;

import java.util.LinkedList;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantLock;

import org.apache.log4j.Logger;

import com.google.protobuf.ByteString;

import cn.com.sparkle.paxos.AccountBook;
import cn.com.sparkle.paxos.ClusterState;
import cn.com.sparkle.paxos.Configuration;
import cn.com.sparkle.paxos.NodesCollection;
//import cn.com.sparkle.paxos.client.net.message.AddRequest;
import cn.com.sparkle.paxos.event.EventsManager;
import cn.com.sparkle.paxos.event.InstanceExecuteEvent;
import cn.com.sparkle.paxos.event.InstanceExecuteMaxPackageSizeEvent;
import cn.com.sparkle.paxos.event.InstancePaxosEvent;
import cn.com.sparkle.paxos.event.MasterChangePosEvent;
import cn.com.sparkle.paxos.event.SpeedControlEvent;
import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages.AddRequest;
import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages.LookUpLatestInstanceIdRequest;
import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages.MessagePackage;
import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages.SuccessfulRecord;
import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages.ValueType;
//import cn.com.sparkle.paxos.model.Id;
//import cn.com.sparkle.paxos.model.Value;
import cn.com.sparkle.paxos.net.MajorSender;
import cn.com.sparkle.paxos.net.NetNode;
import cn.com.sparkle.paxos.net.callback.LookUpLatestInstanceIdCallBack;
//import cn.com.sparkle.paxos.net.message.LookUpLatestInstanceIdRequest;
//import cn.com.sparkle.paxos.net.message.MessagePackage;
import cn.com.sparkle.paxos.paxosinstance.InstancePaxosInstance;
//import cn.com.sparkle.paxos.stablestorage.model.SuccessfulRecord;
import cn.com.sparkle.raptor.core.buff.BuffPool;
import cn.com.sparkle.raptor.core.protocol.MultiThreadProtecolHandler.ProtocolHandlerIoSession;
import cn.com.sparkle.raptor.core.protocol.Protocol;
import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages.Value;
import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages.Id;

public class AddRequestDealer {

	private final static Logger logger = Logger
			.getLogger(AddRequestDealer.class);

	// public final static int BASIC_TCP_PACKAGE_BYTE_SIZE = 512 * 1024;

	// public final static int ALL_TCP_PACKAGE_BYTE_SIZE = 4 *
	// BASIC_TCP_PACKAGE_BYTE_SIZE;

	public final static int MAX_TCP_PACKAGE_SIZE = 1024 * 1024 * 10;

	public final static int MIN_TCP_PACKAGE_SIZE = 1024;

	public final static int MAX_EXECUTING_INSTANCE_NUM = 100;

	public final static int MAX_SESSION_WAIT_QUEUE_SIZE = 3;

	private int executable_instances_num = MAX_EXECUTING_INSTANCE_NUM;

	public volatile int curTcpPackageByteSize = MIN_TCP_PACKAGE_SIZE;

	private LinkedList<AddRequestPackage> requestQueue = new LinkedList<AddRequestPackage>();

	private ClusterState cState;

	private String selfAddress;

	private AccountBook aBook;

	// public volatile int maxInstanceNum;
	private int executingInstanceNum = 0;

	private int quorum;

	private boolean isMaster = false;

	private boolean isOpend = false;

	// private long highSpeedInstanceIdthreshold;// when instanceId more than
	// this
	// value,the master entry the
	// type of high speed that
	// MAX_EXECUTING_INSTANCE_NUM
	// instances will be executed
	// concurrently.

	private ReentrantLock lock = new ReentrantLock();

	private long selfIncreaseInstanceId;

	private EventsManager eventsManager;

	// private boolean isExistManagerCommand = false;

	private int nullCommandCount = 0;// record the num of null command in
										// waitQueue

	public AddRequestDealer(final ClusterState cState,
			final Configuration conf, AccountBook aBook,
			EventsManager aeventsManager) {
		super();
		this.cState = cState;
		this.selfAddress = conf.getSelfAddress();
		this.aBook = aBook;
		this.quorum = conf.getSenators().size() / 2 + 1;
		this.eventsManager = aeventsManager;
		eventsManager.registerEvent(new InstanceExecuteEvent() {
			@Override
			public void eventExecuted(SuccessfulRecord record) {
				if (record.getV().getType() == ValueType.TYPE_ADMIN) {
					// this is management of senator/follower command
					try {
						lock.lock();
						if (isOpend) {// the position of this server must be
										// master.
							reStartDeal();
						}
					} finally {
						lock.unlock();
					}
				}
			}
		});
		eventsManager.registerEvent(new InstancePaxosEvent() {
			int debugCount = 0;

			@Override
			public void success(InstancePaxosInstance instance, Value value) {
				try {
					logger.debug("instance " + instance.getInstanceId()
							+ " succeed!");
					lock.lock();
					if (instance.getAddRequestPackages().get(0)
							.getValueByteSize() == 0) {
						logger.debug("null command successed");
						--nullCommandCount;
					} else if (instance.getAddRequestPackages().get(0)
							.isManageCommand()) {
						// this is admin senator/follower command,the action
						// must be reacted in executeEvent.
						// So nothing to do.
					} else {
						if (instance.getWantAssginValue() != value) {
							for (AddRequestPackage arp : instance
									.getAddRequestPackages()) {
								requestQueue.addFirst(arp);
							}
						} else {
							for (AddRequestPackage arp : instance
									.getAddRequestPackages()) {

								DealState ds = (DealState) arp.getSession().customAttachment;
								AddRequestPackage waitArp = ds.pollFirst();
								if (waitArp != null) {
									requestQueue.addLast(waitArp);
								} else {
									ds.isDealing = false;
								}
								debugCount++;
//								logger.debug("deal request package"
//										+ debugCount + " ds.waitsize"
//										+ ds.size());
							}
						}
					}
					--executingInstanceNum;
					executeNeed();
				} finally {
					lock.unlock();
				}
			}

			@Override
			public void fail(InstancePaxosInstance instance, Id.Builder id,
					long refuseId) {

				try {
					lock.lock();
					if (!isOpend) {
						// stop deal process of this instance and close tcp
						// conntion
						--executingInstanceNum;
						for (AddRequestPackage arp : instance
								.getAddRequestPackages()) {
							arp.closeConnection();
						}

						if (instance.getAddRequestPackages().get(0)
								.getValueByteSize() == 0) {
							// this instance is null command
							--nullCommandCount;
						}
					} else if (refuseId == Long.MAX_VALUE) {
						// The max value of Long indicates the remote peer has
						// succeeded a instance with the same instanceId and the
						// peer has discarded the logs about this instance,so
						// the
						// process of this instance will be stopped.

						if (instance.getAddRequestPackages().get(0)
								.getValueByteSize() != 0) {
							// this instance is not a null command
							for (AddRequestPackage arp : instance
									.getAddRequestPackages()) {
								requestQueue.addFirst(arp);
							}
						} else {
							
							// this instance is null command
							--nullCommandCount;
							logger.debug("this instanceId has succeeded! instanceId : " + instance.getInstanceId() + " nullCommandNum:"  + nullCommandCount);
						}
						--executingInstanceNum;
						executeNeed();
					}else {
						//Long.MAX_VALUE - 1 indicates some file is damaged , and the master can't study from others,and not enough node can provide this instance has succeeded
						if(refuseId != Long.MAX_VALUE - 1){ 
							id.setIncreaseId(refuseId + 1);
						}
						while (true) {
							Set<NetNode> activeNode = cState.getSenators()
									.getActiveNodes();
							if (activeNode.size() >= instance.getQuorum()) {
								instance.resetActiveNode(cState.getSenators()
										.getActiveNodes());
								instance.activate();
								break;
							} else {
								try {
									Thread.sleep(1000);
								} catch (InterruptedException e) {
									e.printStackTrace();
								}
							}
						}
					}

				} finally {
					lock.unlock();
				}
			}

			@Override
			public void start(InstancePaxosInstance instance) {
				// nothing to do
			}
		});

		eventsManager.registerEvent(new MasterChangePosEvent() {
			@Override
			public void masterChange(String address) {
				// nothing to do
			}

			@Override
			public void lostPos() {
				try {
					lock.lock();
					isMaster = false;
					isOpend = false;
					// clear queue of waiting to deal
					for (AddRequestPackage arp : requestQueue) {
						if (arp != null) {
							arp.closeConnection();
						}
					}
					requestQueue.clear();
				} finally {
					lock.unlock();
				}
			}

			@Override
			public void getMasterPos() {
				try {
					lock.lock();
					isMaster = true;
					reStartDeal();
				} finally {
					lock.unlock();
				}
			}
		});

		eventsManager.registerEvent(new SpeedControlEvent() {
			@Override
			public void suggestMaxPackageSize(int suggestSize) {
				suggestSize = Math.max(
						Math.min(suggestSize, MAX_TCP_PACKAGE_SIZE),
						MIN_TCP_PACKAGE_SIZE);
				if (suggestSize != curTcpPackageByteSize) {
					curTcpPackageByteSize = suggestSize;
					InstanceExecuteMaxPackageSizeEvent
							.doMaxPackageSizeChangeEvent(eventsManager,
									curTcpPackageByteSize);
				}
			}
		});
	}

	private void reStartDeal() {
		selfIncreaseInstanceId = aBook.getLastCanExecutableInstanceId();
		// get latest instanceId from others of senators
		NodesCollection senators = cState.getSenators();
		Set<NetNode> activeNodes = senators.getActiveNodes();
		LookUpLatestInstanceIdCallBack callback = new LookUpLatestInstanceIdCallBack(
				quorum, activeNodes.size()) {
			@Override
			public void finish(long instanceId) {
				try {
					lock.lock();
					logger.debug("LookUpLatestInstanceIdCallBack result:"
							+ instanceId);
					if (isMaster) {
						if (instanceId == -2) {
							// manifest the master maybe has lost his
							// position,need
							// to notify lost master
							cState.lostAuthorizationOfMaster();
						} else {
							isOpend = true;
							// highSpeedInstanceIdthreshold = instanceId + 1;
							// insert null operation command
							AddRequest nullRequest = AddRequest.newBuilder()
									.setIsManagerCommand(true)
									.setValue(ByteString.EMPTY)
									.setTransportMaster(true).build();
							MessagePackage.Builder mpBuilder = MessagePackage
									.newBuilder();
							mpBuilder.setId(-1).setIsLast(true)
									.setAddRequest(nullRequest);
							AddRequestPackage nullRequestPackage = new AddRequestPackage(
									mpBuilder.build(), null);
							// When one null command has not been finished,the
							// num of null command may be one less than the
							// count needed.
							// So we need make this count add one.So that the
							// num is more than the count needed,when the server
							// get the position of master.
							int newNullCommandCount = (int) (instanceId - selfIncreaseInstanceId);
							logger.debug("insert " + newNullCommandCount
									+ " null command,current count is "
									+ nullCommandCount);
							for (int i = 0; i < newNullCommandCount
									- nullCommandCount; i++) {
								requestQueue.add(nullRequestPackage);
								executeNeed();
							}
							nullCommandCount = newNullCommandCount;
							InstanceExecuteMaxPackageSizeEvent
									.doMaxPackageSizeChangeEvent(eventsManager,
											curTcpPackageByteSize);
						}
					}
				} finally {
					lock.unlock();
				}

			}
		};
		MajorSender.sendToMajor(activeNodes,
				LookUpLatestInstanceIdRequest.getDefaultInstance(), callback);
	}

	private int debugCount = 0;
	private int addCount = 0;

	public void add(MessagePackage messagePackage,
			ProtocolHandlerIoSession session) {
		try {
			lock.lock();
			if ((addCount++) % 1000 == 0) {
				logger.debug("add count " + (addCount));
			}
			if (!isOpend) {
				session.closeSocketChannel();
				return;
			}
			if (session.customAttachment == null) {
				session.customAttachment = new DealState(lock,
						MAX_SESSION_WAIT_QUEUE_SIZE);
			}
			DealState ds = (DealState) session.customAttachment;

			if (ds.isDealing) {
				AddRequest request = messagePackage.getAddRequest();
				// wait and merge
				AddRequestPackage lastPackage = ds.peekLast();
				if (lastPackage != null
						&& lastPackage.getValueByteSize() < curTcpPackageByteSize
						&& !request.getIsManagerCommand()
						&& !lastPackage.isManageCommand()) {
					int before = lastPackage.valueList.size();
					lastPackage.addRequest(messagePackage);
//					logger.debug("merge  before" + before + "  after" + lastPackage.valueList.size());
				} else {

					ds.add(new AddRequestPackage(messagePackage, session));
//					logger.debug("wait add package:" + (++debugCount)
//							+ "waitSize:" + ds.size());
				}
			} else {
				// no wait
				requestQueue.addLast(new AddRequestPackage(messagePackage,
						session));
//				logger.debug("direct add package:" + (++debugCount));
				ds.isDealing = true;
			}
			executeNeed();
		} finally {
			lock.unlock();
		}

	}

	public void executeNeed() {
		try {
			lock.lock();

			int needExecute = executable_instances_num - executingInstanceNum;

			if (executingInstanceNum == 0) {
				executable_instances_num = MAX_EXECUTING_INSTANCE_NUM; // restore
			}

			for (int i = 0; i < needExecute && requestQueue.size() > 0; ++i) {
				int curByteCount = 0;
				LinkedList<AddRequestPackage> list = new LinkedList<AddRequestPackage>();
				// merge message from all client
				do {
					AddRequestPackage arp = requestQueue.removeFirst();
					if (arp.isManageCommand()) {
						executable_instances_num = 1;// limit executing number
						needExecute = 0; // stop cycle
						if (executingInstanceNum > 0) {
							requestQueue.addFirst(arp);// restore queue
							break;
						}
						curByteCount += MAX_TCP_PACKAGE_SIZE;
					}

					// for null command,there is not DealState,and send a list
					// of size equal 0 indicate a null command
					if (arp.getValueByteSize() != 0) {
						DealState ds = (DealState) arp.getSession().customAttachment;
						ds.isDealing = true;
						curByteCount += arp.getValueByteSize();
					}
					list.add(arp);
//					logger.debug("requestQueue.size()" + requestQueue.size()
//							+ " curTcpPackageByteSize" + curTcpPackageByteSize);
				} while (requestQueue.size() > 0
						&& curByteCount
								+ requestQueue.peek().getValueByteSize() < curTcpPackageByteSize);
				if (list.size() != 0) {
					// assign a id
					++selfIncreaseInstanceId;
					if (selfIncreaseInstanceId < aBook
							.getLastCanExecutableInstanceId()) {
						selfIncreaseInstanceId = aBook
								.getLastCanExecutableInstanceId() + 1;
					}
					Id.Builder idBuilder = Id.newBuilder().setIncreaseId(0)
							.setAddress(selfAddress);
					while (true) {
						Set<NetNode> activeNode = cState.getSenators()
								.getActiveNodes();
						if (activeNode.size() >= quorum) {
							InstancePaxosInstance paxosInstance = new InstancePaxosInstance(
									activeNode, quorum, selfIncreaseInstanceId,
									idBuilder, aBook, list, eventsManager,
									selfAddress);
							paxosInstance.activate();
							break;
						}
						try {
							Thread.sleep(100);
						} catch (InterruptedException e) {
							e.printStackTrace();
						}
					}

					++executingInstanceNum;
				}
			}
		} finally {
			lock.unlock();
		}
	}

}
